package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/12 23:54
 */
public class Flink19_UDTAF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });


        Table table = tableEnv.fromDataStream(sensorDS);


        // 先注册
        tableEnv.createTemporaryFunction("top2", Top2.class);

        // TableAPI
        table
                .groupBy($("id"))
                .flatAggregate(call("top2", $("vc")).as("rank", "top"))
                .select($("id"), $("rank"), $("top"))
                .execute()
                .print();


    }

    public static class Top2 extends TableAggregateFunction<Tuple2<String, Integer>, VcTop2> {


        @Override
        public VcTop2 createAccumulator() {
            return new VcTop2(Integer.MIN_VALUE, Integer.MIN_VALUE);
        }

        public void accumulate(VcTop2 acc, Integer value) {
            if (value > acc.getTopOne()) {
                acc.setTopTwo(acc.getTopOne());
                acc.setTopOne(value);
            } else if (value > acc.getTopTwo()) {
                acc.setTopTwo(value);
            }

        }

        public void emitValue(VcTop2 acc, Collector<Tuple2<String, Integer>> out) {
            out.collect(Tuple2.of("Top1", acc.getTopOne()));
            if (acc.getTopTwo() > Integer.MIN_VALUE) {
                out.collect(Tuple2.of("Top2", acc.getTopTwo()));
            }
        }
    }
}
